package org.artifact.core.plugin.rpc;

import cn.hutool.core.annotation.AnnotationUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.artifact.core.lang.AbstractPlugin;
import org.artifact.core.plugin.rpc.client.RpcClient;
import org.artifact.core.plugin.rpc.client.RpcProxy;
import org.artifact.core.plugin.rpc.codec.RpcDecoder;
import org.artifact.core.plugin.rpc.codec.RpcEncoder;
import org.artifact.core.plugin.rpc.pojo.RpcRequest;
import org.artifact.core.plugin.rpc.pojo.RpcResponse;
import org.artifact.core.plugin.rpc.registry.ServiceDiscovery;
import org.artifact.core.plugin.rpc.registry.ServiceRegistry;
import org.artifact.core.plugin.rpc.registry.zookeeper.ZooKeeperServiceDiscovery;
import org.artifact.core.plugin.rpc.registry.zookeeper.ZooKeeperServiceRegistry;
import org.artifact.core.plugin.rpc.server.RpcServerHandler;
import org.artifact.core.plugin.rpc.server.RpcService;
import org.artifact.core.plugin.zookeeper.StateListener;
import org.artifact.core.plugin.zookeeper.ZooKeeperPlugin;
import org.artifact.core.server.ServerConfig;
import org.artifact.core.server.ServerContext;
import org.artifact.core.server.ServerPlugins;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

public class RpcPlugin extends AbstractPlugin implements StateListener {

    private static final Log log = LogFactory.get();

    private EventLoopGroup bossGroup;

    private EventLoopGroup workerGroup;

    private ServerBootstrap bootstrap;

    private String serviceAddress;

    private String scanPackage;

    private RpcProxy rpcProxy;

    public RpcProxy getRpcProxy() {
        return rpcProxy;
    }

    private ServiceRegistry serviceRegistry;

    private ServiceDiscovery serviceDiscovery;

    private String serviceSuffix;

    public RpcPlugin() {
        Map config = ServerConfig.me().deepGet("plugins.rpc");

        // 创建rpc服务
        serviceAddress = MapUtil.getStr(config, "server.address");
        // 扫描包
        scanPackage = MapUtil.getStr(config, "service.scanPackage");
        if (StrUtil.isEmpty(scanPackage)) {
            scanPackage = ServerContext.me().rootPackage();
        }
        // 服务器后缀
        serviceSuffix = MapUtil.getStr(config, "suffix");

        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();

        // 创建并初始化 Netty 服务端 Bootstrap 对象
        bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new RpcDecoder(RpcRequest.class)); // 解码 RPC 请求
                pipeline.addLast(new RpcEncoder(RpcResponse.class)); // 编码 RPC 响应
                pipeline.addLast(new RpcServerHandler()); // 处理 RPC 请求
            }
        });
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
    }

    @Override
    public boolean start() {
        // 注册中心
        serviceRegistry = new ZooKeeperServiceRegistry();
        serviceDiscovery = new ZooKeeperServiceDiscovery();
        rpcProxy = new RpcProxy(serviceDiscovery);

        // 获取 RPC 服务器的 IP 地址与端口号
        String[] addressArray = StrUtil.split(serviceAddress, ":");
        String ip = addressArray[0];
        int port = Integer.parseInt(addressArray[1]);
        try {
            bootstrap.bind(ip, port).sync();
            log.info("RPC Server Start on {}", port);
        } catch (Exception e) {
            log.error(e);
        }

        // 注册 RPC 服务地址
        if (serviceRegistry != null) {
            register(serviceRegistry, serviceSuffix);
            ServerPlugins.me().getPlugin(ZooKeeperPlugin.class).addStateListener(this);
        }
        return true;
    }

    @Override
    public boolean stop() {
        try {
            // 停止客户端
            for (Entry<String, RpcClient> e : RpcProxy.clients.entrySet()) {
                e.getValue().stop();
            }
            bossGroup.shutdownGracefully().sync();
            workerGroup.shutdownGracefully().sync();
            log.info("RPC Server Stop Success!");
        } catch (InterruptedException e) {
            log.error(e);
        }
        return true;
    }

    private void register(ServiceRegistry serviceRegistry, String serviceVersion) {
        Set<Class<?>> clazzs = ClassUtil.scanPackageByAnnotation(scanPackage, RpcService.class);
        for (Class<?> clazz : clazzs) {
            if (ClassUtil.isNormalClass(clazz)) {
                RpcService rpcService = AnnotationUtil.getAnnotation(clazz, RpcService.class);
                String serviceName = rpcService.value().getName();
                if (StrUtil.isNotEmpty(serviceVersion)) {
                    serviceName += "-" + serviceVersion;
                }
                serviceRegistry.register(serviceName, serviceAddress);
            }
        }


    }

    @Override
    public void stateChanged(int stat) {
        if (stat == StateListener.RECONNECTED) {
            register(serviceRegistry, serviceSuffix);
        }
    }
}
